package com.yzh.dxpro.p1.clear;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.yzh.dxpro.p4.utils.TMatcher;
import com.yzh.dxpro.p4.utils.TProperties;

public class DxClearMapper extends Mapper<LongWritable,Text,NullWritable,Text>{

	@Override
	protected void map(LongWritable key, Text value,
			Mapper<LongWritable, Text, NullWritable, Text>.Context context)
			throws IOException, InterruptedException {
		//读取原始数据，然后对value中的数据进行分割处理，并验证数据进行检验获得结果
		String regex=TProperties.getValue("filesplit");
		//15位
		String[] strs=value.toString().split(regex);
		//检验
		int fileLength=Integer.parseInt(TProperties.getValue("filelength"));
		if (strs.length==fileLength
				&& !"".equals(strs[14])  //split分割的数组不会出现null对象的，因为null不存在分割的问题
				&& !"http://".equals(strs[14])
				&& !"https://".equals(strs[14])
				&& !(strs[14].toLowerCase()).matches(TProperties
						.getValue("fileclear"))) {
			
			//提取域名
			String url=strs[14];
			StringBuffer sb = new StringBuffer();
			//为没有用http://和https://的url地址补上http://
			if (!url.startsWith("http://") && !url.startsWith("https://")) {
				url = "http://" + url;
			}
			// 获取域名
			String domain = url.split("/", -1)[2];
			// 去除端口
			if (domain.indexOf(":") >= 0) {  
				domain = domain.split("\\:", -1)[0];
			}
			//组合结果
			// 用户号码|一级域名|URL地址
			sb.append(strs[1]).append(TProperties.getValue("outfilesplit"))
					.append(TMatcher.getDomain(domain)).append(TProperties.getValue("outfilesplit"))
					.append(url);
			//输出结果
			Text ovalue = new Text(sb.toString());
			context.write(NullWritable.get(), ovalue);
		}
		
	}

}
